package ipl;

import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Hashtable;
import java.util.StringTokenizer;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class SecondMapper extends
		Mapper<LongWritable, Text, Text, NullWritable> {

	private ArrayList<Integer> scanList;
	private Hashtable<Integer, Integer> keyIndex;

	protected void setup(Context context) throws IOException,
			InterruptedException {
		Path[] paths = context.getLocalCacheFiles();
		File f = new File(paths[0].toString());
		// Files.readAllLines(new
		// java.nio.file.Path(paths[0].toString()),StandardCharsets.UTF_8);
		Charset charset = StandardCharsets.UTF_8;
		String partialSumList = "";
		try (BufferedReader reader = Files.newBufferedReader(f.toPath(),
				charset)) {

			String line = null;
			while ((line = reader.readLine()) != null) {
				partialSumList += line;
			}
			partialSumList.trim();
		} catch (IOException x) {
			System.err.format("IOException: %s%n", x);
		}

		// String scanSum = FileSystem.get(context.getConfiguration()).open(new
		// Path(paths[0].toString())).readLine().trim(); // not very good
		// scanSum = scanSum.split("\t")[1];

		if (partialSumList.length() > 0) {
			keyIndex = new Hashtable<Integer, Integer>();
			String[] tmp = partialSumList.split(";");
			scanList = new ArrayList<Integer>();

			scanList.add(0);
			int sum = 0;
			int counter = 0;
			for (String val : tmp) {
				if (val != null && val.length() > 0) {
					int key = Integer.parseInt(val.split("=")[0]);
					int parSum = Integer.parseInt(val.split("=")[1]);
					keyIndex.put(key, counter);
					sum += parSum;
					scanList.add(sum); // add all of them to this list, from 0
										// to last partial-sum
				}
				counter++;
			}
		}

	}
	
	public void map(LongWritable key, Text list, Context context)
			throws IOException, InterruptedException {
		if (scanList.size() > 0) {
			int idx = (int) key.get();
			int sum = scanList.get(keyIndex.get(idx));
			StringBuffer newList = new StringBuffer(512);
			StringTokenizer itr = new StringTokenizer(list.toString().trim());
			while (itr.hasMoreTokens()) {
				newList.append(sum+" ");
				sum += Integer.parseInt(itr.nextToken());
				
			}
			context.write(new Text(newList.toString()), NullWritable.get());
			
//			String[] eles = list.toString().trim().split(" ");
//			
//			String newList = "";
//			// prescan
//			
//			for (int i = 0; i < eles.length; i++) {
//				newList += sum + " ";
//				sum += Integer.parseInt(eles[i]);
//			}
//			context.write(new Text(newList), NullWritable.get());
		}

	}
}
